package cn.lsh.spark.streaming;

import org.apache.log4j.lf5.LogLevel;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.StreamingContext;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.dstream.DStream;

public class FileSparkStreaming {
	
	public static void main(String[] args) throws InterruptedException {
		//不需要用JavaReceiverInputDStream接收数据，一个线程就可以了
		JavaStreamingContext jsc = new JavaStreamingContext("local[1]", "FileSparkStreaming", Durations.seconds(5));
		jsc.sparkContext().setLogLevel(LogLevel.WARN.getLabel());
		//可以设置checkpoint，保存Context的同步时保存offset（不然SparkStreaming每次启动都不会从之前的offset开始消费），也可以不设置checkpoint，自己管理offset
		jsc.checkpoint("file:/checkpoint/stream_context");
		//监控文件目录，本地目录或hdfs都可以。SparkStreaming根据间隔时间扫面指定目录，如果发现有新增的文件（和上一次扫描对比），读取新增的文件数据。
		//这里，如果写文件的时间超过了batchDuration，会导致第一次发现新增文件，但文件是空的，第二次就不会再扫该文件。
		// 所以一个文件创建flush数据的时间要小于batchDuration，手动黏贴文件也不行，追加文件内容也不行
		JavaDStream<String> fileDStream = jsc.textFileStream("file:/data");
		fileDStream.foreachRDD(rdd -> rdd.foreach(s -> System.out.println(s)));
		DStream<String> dstream = fileDStream.dstream();
		dstream.print();
		//saveAsTextFiles保存DStream为文件，不管有没有数据都会保存为一个文件
		dstream.saveAsTextFiles("file:/savedata/prefix", "txt");
		jsc.start();
		jsc.awaitTermination();
	}
}
